package com.hownoon.parquet1;

import com.hownoon.Util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.log4j.BasicConfigurator;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;

public class ParquetCollet {
    private FileSystem fileSystem;

    public ParquetCollet(String floderPath) {
        before();
        exec(floderPath);
        after();
    }

    public void exec(String floderPath){
        long sum=0;
        ArrayList<Path> arrayList=getAllPath(floderPath);
        for(int i=0;i<arrayList.size();i++){
            sum=sum+read(arrayList.get(i));
        }
        System.out.println("全部成交单数为:"+sum+"条");
    }

    public void before(){
        BasicConfigurator.configure(); //自动快速地使用缺省Log4j环境
        Configuration configuration=new Configuration();
        //configuration.set("dfs.replication","1");
        try {
            fileSystem= FileSystem.get(URI.create(Util.url),configuration,"hadoop");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public void after(){
        try {
            fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public long read(Path path){
        InputStream inputStream=null;
        InputStreamReader inputStreamReader = null;
        BufferedReader bufferedReader=null;
        long result=0;
        try {
            inputStream=fileSystem.open(path);

            //inputStream=fileSystem.open(new Path("hdfs://172.16.50.51:9820/result/part-r-00000"));
            inputStreamReader=new InputStreamReader(inputStream);
            bufferedReader=new BufferedReader(inputStreamReader);
            result= Long.parseLong(bufferedReader.readLine());
            bufferedReader.close();
            inputStreamReader.close();

        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            IOUtils.closeStream(inputStream);
        }
        return result;
    }

    public ArrayList<Path> getAllPath(String floderPath){
        ArrayList<Path> arrayList=new ArrayList<>();
        try {
            Path path=new Path(floderPath);
            FileStatus fileStatus[]=fileSystem.listStatus(path);
            for(FileStatus fileStatus1:fileStatus){
                if(fileStatus1.isFile()){
                    if(fileStatus1.getPath().getName().indexOf("part")>=0) {
                        arrayList.add(fileStatus1.getPath());
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }finally {

        }
        return arrayList;
    }

}
